package internal;

import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;

import static java.util.stream.Collectors.toList;

/**
 *  事件的调度,事件解耦真正执行的地方
 * Created by DJJ on 2019/3/28.
 */
public class MyDispatcher {

    private final Executor executorService;

    private final MyEventExceptionHandler exceptionHandler;

    public static final Executor SEQ_EXECUTOR_SERVICE = SeqExecutorService.INSTANCE;

    public static final Executor PRE_THREAD_EXECUTOR_SERVICE = PreThreadExecutorService.INSTANCE;
    //用于使用线程池环境下关闭线程池判断
    private CountDownLatch countDownLatch =null;

    public MyDispatcher(Executor executorService, MyEventExceptionHandler exceptionHandler) {
        this.executorService = executorService;
        this.exceptionHandler = exceptionHandler;
    }
    //提供工厂方法
    static MyDispatcher newDispatcher(MyEventExceptionHandler exceptionHandler, Executor executor)
    {
        return new MyDispatcher(executor, exceptionHandler);
    }

    /**
     *
     * @param bus
     * @param myRegistry
     * @param event 要推送的消息类型,string,int ...
     * @param topic
     */
    public void dispatch(Bus bus,MyRegistry myRegistry,Object event,String topic){
        ConcurrentLinkedDeque<MySubscriber> subscribers = myRegistry.scanSubscriber(topic);
        if (subscribers==null){
            if (exceptionHandler != null) {
                exceptionHandler.handle(new IllegalArgumentException("The topic " + topic + " not bind yet"),
                        new BaseMyEventContext(bus.getBusName(),null,event) {
                        });
            }
            return;
        }
        List<MySubscriber> collect = subscribers.stream().filter(sub -> !sub.isDisable()).filter(mySubscriber -> {
            Method subscribeMethod = mySubscriber.getSubscribeMethod();
            Class<?> parameterType = subscribeMethod.getParameterTypes()[0];
            //校验事件消息订阅的方法的参数类型和传入的post的参数类型相同
            return parameterType.isAssignableFrom(event.getClass());
        }).collect(toList());
        countDownLatch= new CountDownLatch(collect.size());
        collect.stream().forEach(subscriber->realInvokeSubscribe(subscriber,event,bus,countDownLatch));

    }

    /**
     *
     * @param subscriber 封装的订阅对象
     * @param event  要post 消息对象
     *
     */
    public void realInvokeSubscribe(MySubscriber subscriber,Object event,Bus bus,CountDownLatch countDownLatch){
        Method method = subscriber.getSubscribeMethod();//要推送的对象方法
        Object object = subscriber.getSubscribeObject();//推送的对象
        executorService.execute(()->
                {
                    try {
                        method.invoke(object,event);
                    } catch (Exception e) {
                       if (exceptionHandler!=null){
                           exceptionHandler.handle(e,new BaseMyEventContext(bus.getBusName(),subscriber,event));
                       }
                    }finally {
                        countDownLatch.countDown();
                    }
                }
                );
    }
    public void close() throws InterruptedException {
        countDownLatch.await();
        if (executorService instanceof ExecutorService)
          ((ExecutorService) executorService).shutdown();

    }

    /**
     * 串行的执行事件消息
     */
    private static class SeqExecutorService implements Executor{
        private final  static SeqExecutorService INSTANCE=new SeqExecutorService();
        @Override
        public void execute(Runnable command) {
            command.run();
        }
    }

    /**
     * 并行的执行事件消息
     */
    private static class PreThreadExecutorService implements Executor{
        private final  static PreThreadExecutorService INSTANCE=new PreThreadExecutorService();
        @Override
        public void execute(Runnable command) {
            new Thread(command).start();
        }
    }

    /**
     * 事件消息上下文
     */
    private static class BaseMyEventContext implements MyEventContext{

        private final String eventBusName;

        private final MySubscriber subscriber;

        private final Object event;

        public BaseMyEventContext(String eventBusName, MySubscriber subscriber, Object event) {
            this.eventBusName = eventBusName;
            this.subscriber = subscriber;
            this.event = event;
        }

        @Override
        public String getSource() {
            return this.eventBusName;
        }

        @Override
        public Object getSubscriber() {
            return subscriber.getSubscribeObject();
        }

        @Override
        public Method getSubscribe() {
            return subscriber.getSubscribeMethod();
        }

        @Override
        public Object getEvent() {
            return this.event;
        }
    }
}
